package cn.doitedu.demo7;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;
import java.util.Map;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/7
 * @Desc: 学大数据，上多易教育
 * 实现： 可以动态注入、下线、更新规则
 * 当然，前提是，这些规则，都是系统已经内置的 规则模型下的规则
 **/
@Slf4j
public class MainEntry3_BroadcastState {

    //private static final Logger log = LoggerFactory.getLogger(MainEntry.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setStateBackend(new HashMapStateBackend());


        // 构造cdc连接器，实时抓取 规则元数据表中的数据变更
        MySqlSource<String> ruleMetaCdcSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit46") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit46.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        // 解析cdc抓取的规则元数据变更
        DataStreamSource<String> cdcStream = env.fromSource(ruleMetaCdcSource, WatermarkStrategy.noWatermarks(), "cdc");

        // 对抓取到的ruleMeta变更原始json数据，进行解析、提取
        SingleOutputStreamOperator<RuleMeta> ruleMetaBeans = cdcStream.map(json -> {

            JSONObject jsonObject = JSON.parseObject(json);
            String op = jsonObject.getString("op");

            JSONObject dataObj;
            if (op.equals("d")) {
                dataObj = jsonObject.getJSONObject("before");
            } else {
                dataObj = jsonObject.getJSONObject("after");
            }

            byte[] bytes = dataObj.getBytes("pre_select_users");

            String jsonString = dataObj.toJSONString();
            RuleMeta ruleMeta = JSON.parseObject(jsonString, RuleMeta.class);

            // 填充op字段
            ruleMeta.setOp(op);

            // 反序列化 bitmap的字节数组
            byte[] preSelectUsersBytes = ruleMeta.getPre_select_users();
            Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf();

            if (preSelectUsersBytes != null) {
                bitmap.deserialize(ByteBuffer.wrap(preSelectUsersBytes));
                log.info("反序列化得到的bitmap是: {}", bitmap);
            }
            // 填充bitmap字段
            ruleMeta.setPreSelectBitmap(bitmap);

            return ruleMeta;
        });


        // 将规则元数据流，进行广播
        MapStateDescriptor<String, RuleCalculator> desc = new MapStateDescriptor<>("bc_state", String.class, RuleCalculator.class);
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaBeans.broadcast(desc);


        // 构建kafka source来读取数据
        KafkaSource<String> source = KafkaSource
                .<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("user-action-log")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doitedu-ww")
                .setClientIdPrefix("doitedu-cc")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "log-source");

        // 将source读取到的json数据，转换成javaBean数据
        DataStream<UserAction> mapped = stream.map(json -> JSON.parseObject(json, UserAction.class));

        // 按用户分区
        KeyedStream<UserAction, Long> keyedStream = mapped.keyBy(UserAction::getUser_id);


        // 将用户行为流 ， 连接   规则元数据广播流
        keyedStream.connect(ruleMetaBroadcastStream)
                .process(new KeyedBroadcastProcessFunction<Long, UserAction, RuleMeta, String>() {

                    //HashMap<String, RuleCalculator> ruleCalculatorPool;
                    //MapState<String, RuleCalculator> ruleCalculatorPool;

                    // 最近2分钟的用户行为明细缓存
                    ListState<UserAction> recentActions;

                    // 用户存放已处理过数据的运算机的规则id
                    MapState<String, String> noNewCalculatorIds;

                    GroovyClassLoader groovyClassLoader;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        //ruleCalculatorPool = new HashMap<>();
                        //ruleCalculatorPool = getRuntimeContext().getMapState(new MapStateDescriptor<String, RuleCalculator>("calculator-pool", String.class, RuleCalculator.class));


                        ListStateDescriptor<UserAction> desc = new ListStateDescriptor<>("recent_actions", UserAction.class);
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.minutes(2))
                                .neverReturnExpired()
                                .build();
                        desc.enableTimeToLive(ttlConfig);

                        recentActions = getRuntimeContext().getListState(desc);


                        this.noNewCalculatorIds = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("no_new_calculator_ids", String.class, String.class));


                        this.groovyClassLoader = new GroovyClassLoader();

                    }

                    @Override
                    public void processElement(UserAction currentAction, KeyedBroadcastProcessFunction<Long, UserAction, RuleMeta, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {

                        // 获取广播状态，用来管理运算机对象
                        ReadOnlyBroadcastState<String, RuleCalculator> ruleCalculatorPool = ctx.getBroadcastState(new MapStateDescriptor<String, RuleCalculator>("bc_state", String.class, RuleCalculator.class));


                        // 先把本次的用户行为放入缓存状态
                        recentActions.add(currentAction);


                        // 遍历运算机池
                        for (Map.Entry<String, RuleCalculator> entry : ruleCalculatorPool.immutableEntries()) {

                            // 取到一个规则运算机对象
                            String ruleId = entry.getKey();
                            RuleCalculator ruleCalculator = entry.getValue();

                            // 判断该运算机是否是一个新上线的运算机
                            if (!noNewCalculatorIds.contains(ruleId)) {
                                // 如果是，则要把缓存状态中的最近2分钟历史数据交给这个运算机处理
                                for (UserAction action : recentActions.get()) {
                                    ruleCalculator.calc(action, out);
                                }

                                // 然后将这个新运算加入 “老运算机id集合”
                                noNewCalculatorIds.put(ruleId, "");

                            } else {
                                // 如果否,那就把本次收到的这一条行为交给它处理
                                ruleCalculator.calc(currentAction, out);
                            }
                        }
                    }

                    @Override
                    public void processBroadcastElement(RuleMeta ruleMeta, KeyedBroadcastProcessFunction<Long, UserAction, RuleMeta, String>.Context ctx, Collector<String> out) throws Exception {

                        // 获取广播状态，作为运算机池
                        BroadcastState<String, RuleCalculator> ruleCalculatorPool = ctx.getBroadcastState(new MapStateDescriptor<String, RuleCalculator>("bc_state", String.class, RuleCalculator.class));


                        String op = ruleMeta.getOp();

                        // 如果是元数据删除操作，或 元数据更新操作且更新后的规则状态为0 ，都表示要下线规则
                        if (op.equals("d") || (op.equals("u") && ruleMeta.getRule_status() == 0)) {
                            ruleCalculatorPool.remove(ruleMeta.getRule_id());

                            log.warn("下线了一个规则运算机,rule_id:{}", ruleMeta.getRule_id());

                        }
                        // 如果是初始读或新增了一条规则元数据，且规则的状态为1，表示要上线规则
                        else if ((op.equals("r") || op.equals("c") || op.equals("u")) && ruleMeta.getRule_status() == 1) {
                            // 根据规则元数据，构造运算机对象
                            String ruleModelId = ruleMeta.getRule_model_id();


                            // 从规则元数据中，取出规则对应的模型运算机groovy代码
                            String modelCode = ruleMeta.getModel_code();
                            Class<?> aClass = groovyClassLoader.parseClass(modelCode);

                            RuleCalculator ruleCalculator = (RuleCalculator) aClass.newInstance();


                            // 并利用元数据中的 规则参数，初始化运算机，得到具体规则运算机对象
                            ruleCalculator.init(getRuntimeContext(), ruleMeta);

                            // 将初始化好的运算机对象，放入运算机池
                            ruleCalculatorPool.put(ruleMeta.getRule_id(), ruleCalculator);

                            log.info("上线了一个规则运算机,rule_id:{}", ruleMeta.getRule_id());


                        } else {
                            // TODO :  NOTHING TO DO
                            log.info("收到一个规则元数据，但我什么也没做,rule_id:{}", ruleMeta.getRule_id());
                        }
                    }
                }).print();


        env.execute();


    }

}
